package com.heima.article.stream;

import com.alibaba.cloud.commons.lang.StringUtils;
import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.time.Duration;

@Component
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //1. 订阅主题 拉取消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        //2. 对消息进行转换 key:1dssklafjd value:{l:1,aid:110} -> key:文章id value:  likes:1
        stream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
                return new KeyValue<>(updateArticleMess.getArticleId().toString(), updateArticleMess.getType().name() + ":" + updateArticleMess.getAdd());
            }
        })
                .groupBy((key,value) -> key) //3. 根据文章的id进行分组统计
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10))) //4. 设置时间窗口
                .aggregate(new Initializer<String>() {  //5. 自定义统计规则
                    /**
                     * 再每一个时间窗口开始的时候完善数据的清除和持久化的工作
                     * 每一个时间窗口只执行一次
                     * @return
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     * 每隔消息来的时候都会执行一次
                     * @param key 消息的key -->文章的id
                     * @param value 消息的value --> like:1
                     * @param aggValue 上一次统计的结果，如果是第一个消息 则是初始化的结果
                     * @return
                     *
                     * aggValue: COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                     */
                    @Override
                    public String apply(String key, String value, String aggValue) {
                        if(StringUtils.isEmpty(value)){
                            return aggValue;
                        }

                        //恢复上一次统计的结果
                        String[] strs = aggValue.split(","); //[COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0]
                        int col = 0,com=0,lik=0,vie=0;
                        for (String str : strs) {
                            String[] temp = str.split(":"); //[COLLECTION,0]
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(temp[0])){
                                case COLLECTION:
                                    col = Integer.parseInt(temp[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(temp[1]);
                                    break;
                                case LIKES:
                                    lik = Integer.parseInt(temp[1]);
                                    break;
                                case VIEWS:
                                    vie = Integer.parseInt(temp[1]);
                            }
                        }

                        //解析value的数据
                        String[] values = value.split(":"); //[COLLECTION,1]
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(values[0])){
                            case COLLECTION:
                                col += Integer.parseInt(values[1]);
                                break;
                            case COMMENT:
                                com += Integer.parseInt(values[1]);
                                break;
                            case LIKES:
                                lik += Integer.parseInt(values[1]);
                                break;
                            case VIEWS:
                                vie += Integer.parseInt(values[1]);
                        }

                        //还原统计的数据格式
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        System.out.println("文章的id:"+key);
                        System.out.println("当前时间窗口内的消息处理结果："+formatStr);

                        return formatStr;
                    }
                }, Materialized.as("hot-atricle-stream-count-001"))
                .toStream() //6. 把结果转换未kstream类型的对象
                .map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), formatObj(key.key().toString(), value.toString()));
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC); //7. 把结果发送到新的topic

        return stream;
    }

    /**
     * COLLECTION:1,COMMENT:0,LIKES:8,VIEWS:10
     * @param articleId
     * @param value
     * @return
     */
    private Object formatObj(String articleId, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(articleId));

        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理之后的结果为:{}",JSON.toJSONString(mess));

        return JSON.toJSONString(mess);
    }
}
